package org.my.hadoop.test;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class STjoin {
    public static int time = 0;
    public static class Map extends Mapper<Object, Text, Text, Text>{
       public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
    	   String childname;
    	   String parentname;
    	   String relationType;
    	   
    	   StringTokenizer line = new StringTokenizer(value.toString());
    	   System.out.println("*****>"+value.toString()+"<**************");
    	   String[] values = {line.nextToken(),line.nextToken()};
    	   if(!values[0].equals("child")){
    		   childname = values[0];
    		   parentname = values[1];
    		   
    		   relationType="1";
    		   context.write(new Text(parentname), new Text(relationType+"+"+childname+"+"+parentname));
    		   
    		   relationType = "2";
    		   context.write(new Text(childname), new Text(relationType+"+"+childname+"+"+parentname));
    	   }
       }	
    
    }
    
    
    public static class Reduce extends Reducer<Text, Text, Text, Text>{
    	public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
    		if(time==0){
    			context.write(new Text("grandchild"), new Text("grandparent"));
    			time++;
    		}
    		int grandchildnum = 0;
    		String[] grandchild = new String[10];
    		int grandparentnum = 0;
    		String[] grandparent = new String[10];
    		for(Text val:values){
    			String record =val.toString();
    			int len = record.length();
    		    if(len == 0){
    		    	continue;
    		    }
    		    StringTokenizer line = new StringTokenizer(record,"+");
    		    String relationType = line.nextToken();
    		    String childname = line.nextToken();
    		    String parentname = line.nextToken();
    		    
    		    if(relationType.equals("1")){
    		    	grandchild[grandchildnum++] = childname;
    		    }else{
    		    	grandparent[grandparentnum++] = parentname;
    		    }
    		    
    		}
    		
    		if(grandchildnum!=0&&grandparentnum!=0){
    			for(int m = 0;m<grandchildnum;m++){
    				for(int n=0;n<grandparentnum;n++){
    					context.write(new Text(grandchild[m]), new Text(grandparent[n]));
    				}
    			}
    		}
    	}
    }
    
    public static void main(String[] args) throws Exception{
		Configuration conf = new Configuration();
		if(args.length<2){
			System.err.println("Usage:STjoin <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf,"single table join");
		job.setJarByClass(STjoin.class);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		int i=0;
		for(;i<args.length-1;i++){
			FileInputFormat.addInputPath(job, new Path(args[i]));
		}
		FileOutputFormat.setOutputPath(job, new Path(args[i]));
		System.exit(job.waitForCompletion(true)?0:1);
		
	}
}
